/********************************************************************
	file name:	hy_thread_base.h
	created:	2012/03/01 13:30
	author:		supercaco@sina.com
	purpose:	thread
*********************************************************************/
#ifndef _HY_QUEUE_THREAD_H_
#define _HY_QUEUE_THREAD_H_


#include "hy_thread.h"
#include "hy_runnable.h"

class hy_thread_base : public hy_noncopyable
{
public:
	hy_thread_base():thread_(NULL)
	{
		
	}
	virtual ~hy_thread_base()
	{
		wait();
	}

	virtual void	stop() = 0;
	virtual void	run()  = 0;


	hy_err_t		start()
	{
		if (thread_ == NULL)
		{
			void* param = NULL;
			thread_ = new hy_thread(bind_thread(&hy_thread_base::_run,this,param));
		}
		return thread_ == NULL ? hy_error : hy_right;
	}

	//wait for thread exit
	void			wait()
	{
		if (thread_ != NULL)
		{
			thread_->join();
			delete thread_;
			thread_ = NULL;
		}
	}

	bool			is_run()
	{
		if (thread_ != NULL)
		{
			return thread_->status() == hy_thread_running;
		}
		return false;
	}
protected:
	void _run(void*)
	{
		run();
	}

protected:
	hy_thread*	thread_;
};

//thread with queues 
template<typename Queue>
class hy_thread_with_queues: public hy_thread_base
{
	typedef typename Queue::data_block_t data_block_type;
public:
	hy_thread_with_queues():quit_(false)
	{

	}

	~hy_thread_with_queues()
	{

	}

	void add_queue(data_block_type* data_block)
	{
		queue_.push_back(data_block);
	}

	virtual void do_work(data_block_type* data_block) = 0;

	void	stop()
	{
		quit_ = true;
	}

	void	run()
	{
		Queue work_queue;

		while (true)
		{
			if (queue_.empty())
			{
				if (quit_)
				{
					break;
				}
				hy_sleep(300);
			}
			else
			{
				queue_.swap(work_queue);
	
				while(!work_queue.empty())
				{
					do_work(work_queue.pop_front());
				}
				work_queue.clear();
			}
		}
	}

private:
	Queue				queue_;
	volatile bool		quit_;	
};

template<typename Queue>
class hy_thread_with_queues_heart: public hy_thread_base
{
	typedef typename Queue::data_block_t data_block_type;
public:
	hy_thread_with_queues_heart():quit_(false),heart_(true)
	{

	}

	~hy_thread_with_queues_heart()
	{

	}

	void add_queue(data_block_type* data_block)
	{
		queue_.push_back(data_block);
	}

	virtual void do_work(data_block_type* data_block) = 0;

	void	stop()
	{
		quit_ = true;
	}

	void	run()
	{
		Queue work_queue;

		while (1)
		{
			if (queue_.empty() && !heart_)
			{
				if (quit_)
				{
					break;
				}
				hy_os::hy_sleep(100);
				heart_ = true;
			}
			else if (heart_)
			{
				heart_ = false;
				do_work(NULL);
			}
			else
			{
				queue_.swap(work_queue);

				while(!work_queue.empty())
				{
					if (!heart_)
					{
						do_work(work_queue.pop_front());
						heart_ = true;
					}
					else
					{
						do_work(NULL);
						heart_ = false;						
					}
				}
				work_queue.clear();
			}
		}
	}

private:
	Queue				queue_;
	volatile bool		quit_;	
	volatile bool		heart_;
};

#endif //_HY_QUEUE_THREAD_H_

